/*
 * Copyright (c) 2024 Huawei Technologies Co.,Ltd.
 *
 * openGauss is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *
 * http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 * -------------------------------------------------------------------------
 *
 * htap_internal.cpp
 * Delta tables for HTAP
 *
 * IDENTIFICATION
 * src/gausskernel/storage/htap/htap_internal.cpp
 *
 * -------------------------------------------------------------------------
 */

#include <ostream>
#include <istream>
#include <iomanip>
#include <pthread.h>
#include <cstring>

#include "access/tableam.h"
#include "global.h"
#include "mot_engine.h"
#include "thread_id.h"
#include "index_iterator.h"
#include "htap_debug.h"
#include "htap_logger.h"
#include "htap_helpers.h"
#include "htap_internal.h"
#include "htap_statistics.h"
#include "table.h"
#include "htap_delta_table.h"
#include "commands/dbcommands.h"
#include "access/imcs/imcu_cache_mgr.h"
#include "access/imcs/imcs_ctlg.h"
#include "../mot/fdw_adapter/gaussdb_config_loader.h"
#include "htap_csn_manager.h"
#include "htap_infra.h"
#include "vl_utils.h"

HTAP_DECLARE_LOGGER(INTERNAL, HTAP)

static MOT::MOTEngine *htapEngine = nullptr;
static GaussdbConfigLoader *htapConfigLoader = nullptr;
static htap::HTAPCSNManager g_snapshotMgr;

static bool htapToTerminate = false;

using namespace MOT;
namespace htap {

bool HTAPAdaptor::m_initialized = false;
bool HTAPAdaptor::m_callbacks_initialized = false;

inline static Numeric MOTNumericToPG(MOT::DecimalSt *d)
{
    NumericVar v;

    v.ndigits = d->m_hdr.m_ndigits;
    v.dscale = d->m_hdr.m_scale;
    v.weight = (int)(int16_t)(d->m_hdr.m_weight);
    v.sign = (d->m_hdr.m_flags & DECIMAL_POSITIVE ?
                                                  NUMERIC_POS :
                                                  (d->m_hdr.m_flags & DECIMAL_NEGATIVE ? NUMERIC_NEG : ((d->m_hdr.m_flags & DECIMAL_NAN) ? DECIMAL_NAN : 0)));
    v.buf = (NumericDigit *)&d->m_round;
    v.digits = (NumericDigit *)d->m_digits;

    return makeNumeric(&v);
}

void HTAPAdaptor::MOTToDatum1(MOT::Column *col, int2 attlen, Oid type, int deltaColIdx, uint8_t *data, bool *isNull,
                 ScalarVector *colVec, uint32_t recIdx)
{
    Datum value;
    Datum *result_val = &value;

    if (deltaColIdx >= 0 && !BITMAP_GET(data, deltaColIdx)) {
        *isNull = true;
        *result_val = PointerGetDatum(nullptr);
        return;
    }

    size_t len = 0;

    *isNull = false;
    switch (type) {
        case VARCHAROID:
        case BPCHAROID:
        case NVARCHAR2OID:
        case TEXTOID:
        case CLOBOID:
        case BYTEAOID: {
            uintptr_t tmp;
            bytea *result;
            col->Unpack(data, &tmp, len);
            if (attlen > 0 && attlen <= 8) {
                result = (bytea *)&colVec->m_vals[recIdx];
            } else {
                result = (bytea *)colVec->m_buf->Allocate(len + VARHDRSZ);
                colVec->m_vals[recIdx] = PointerGetDatum(result);
            }

            if (len > 0) {
                errno_t erc = memcpy_s(VARDATA(result), len, (uint8_t *)tmp, len);
                securec_check(erc, "\0", "\0");
            }
            SET_VARSIZE(result, len + VARHDRSZ);
            break;
        }
        case NUMERICOID: {
            MOT::DecimalSt *d;
            col->Unpack(data, (uintptr_t *)&d, len);
            if (attlen > 0 && attlen <= 8) {
                colVec->m_vals[recIdx] = NumericGetDatum(MOTNumericToPG(d));
            } else {
                colVec->AddVar(NumericGetDatum(MOTNumericToPG(d)), recIdx);
            }
            break;
        }
        default:
            col->Unpack(data, result_val, len);
            if (attlen > 0 && attlen <= 8) {
                colVec->m_vals[recIdx] = *result_val;
            } else {
                colVec->AddVar(*result_val, recIdx);
            }
            break;
    }
}

void HTAPAdaptor::MOTToDatum1(MOT::Column *col, int2 attlen, Oid type, int deltaColIdx, uint8_t *data, bool *isNull,
                 Datum *value_out, size_t *lenOut)
{
    Datum *result_val = value_out;

    Assert(value_out);
    if (deltaColIdx >= 0 && !BITMAP_GET(data, deltaColIdx)) {
        *isNull = true;
        *result_val = PointerGetDatum(nullptr);
        return;
    }

    size_t *len;
    size_t l = 0;
    if (lenOut) {
        len = lenOut;
    } else {
        len = &l;
    }

    *isNull = false;
    switch (type) {
        case VARCHAROID:
        case BPCHAROID:
        case NVARCHAR2OID:
        case TEXTOID:
        case CLOBOID:
        case BYTEAOID: {
            uintptr_t tmp;
            bytea *result;
            col->Unpack(data, &tmp, *len);
            result = (bytea *)palloc0(*len + VARHDRSZ);

            if (*len > 0) {
                errno_t erc = memcpy_s(VARDATA(result), *len, (uint8_t *)tmp, *len);
                securec_check(erc, "\0", "\0");
            }
            SET_VARSIZE(result, *len + VARHDRSZ);
            *result_val = PointerGetDatum(result);
            break;
        }
        case NUMERICOID: {
            MOT::DecimalSt *d;
            col->Unpack(data, (uintptr_t *)&d, *len);
            *result_val = NumericGetDatum(MOTNumericToPG(d));
            break;
        }
        default:
            col->Unpack(data, result_val, *len);
            break;
    }
}

/** @brief Translation function for converting error codes to string format. */
static const char *HTAPErrorCodeTranslate(int errorCode)
{
    return htap::RcToString((htap::RC)errorCode);
}

bool InitHTAPEngine()
{
    HTAP_LOG_INFO("Initializing htap engine");
    if ((htapEngine = MOT::MOTEngine::GetInstance()) != nullptr) {
        HTAP_LOG_INFO("htap engine already initialized");

        // make sure current thread is cleaned up properly when thread pool is enabled
        // avoid throwing errors on failure
        if (!htap::HTAPEnsureSafeThreadAccess()) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to initialize thread-local data");
            return false;
        }
        return true;
    }
    bool addedCfgListener = false;
    do { // instead of goto
        htapEngine = MOT::MOTEngine::CreateInstanceNoInit(g_instance.attr.attr_common.HTAPConfigFileName, 0, nullptr);
        if (htapEngine == nullptr) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to create htap engine");
            break;
        }

        MOT::MOTConfiguration &htapCfg = MOT::GetGlobalConfiguration();
        htapCfg.SetTotalMemoryMb(g_instance.attr.attr_memory.max_process_memory / KILO_BYTE);

        htapConfigLoader = new (std::nothrow) GaussdbConfigLoader();
        if (htapConfigLoader == nullptr) {
            HTAP_REPORT_ERROR(HTAP_ERROR_OOM, "Init HTAP",
                              "Failed to allocate memory for GaussDB/htap configuration loader");
            break;
        }
        HTAP_LOG_TRACE("Adding external configuration loader for GaussDB");
        if (!htapEngine->AddConfigLoader(htapConfigLoader)) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to add GaussDB/htap configuration loader");
            break;
        }
        addedCfgListener = true;

        if (!htapEngine->LoadConfig()) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to load configuration for htap engine");
            break;
        }

        // Check max process memory here - we do it anyway to protect ourselves from miscalculations.
        // Attention: the following values are configured during the call to MOTEngine::LoadConfig() just above
        uint64_t globalMemoryKb = MOT::g_memGlobalCfg.m_maxGlobalMemoryMb * KILO_BYTE;
        uint64_t localMemoryKb = MOT::g_memGlobalCfg.m_maxLocalMemoryMb * KILO_BYTE;
        uint64_t maxReserveMemoryKb = globalMemoryKb + localMemoryKb;

        // check whether the 2GB gap between MOT and envelope is still kept
        if ((g_instance.attr.attr_memory.max_process_memory < (int32)maxReserveMemoryKb) ||
            ((g_instance.attr.attr_memory.max_process_memory - maxReserveMemoryKb) < MIN_DYNAMIC_PROCESS_MEMORY)) {
            // we allow one extreme case: GaussDB is configured to its limit, and zero memory is left for us
            if (maxReserveMemoryKb <= MOT::MOTConfiguration::MOT_MIN_MEMORY_USAGE_MB * KILO_BYTE) {
                HTAP_LOG_WARN("Allowing htap to work in  memory mode");
            } else {
                HTAP_REPORT_ERROR(HTAP_ERROR_INVALID_CFG, "Init HTAP",
                                  "The value of pre-reserved memory for htap engine is not reasonable: "
                                  "Request for a maximum of %" PRIu64 " KB global memory, and %" PRIu64
                                  " KB session memory (total of %" PRIu64 " KB) is invalid since "
                                  "max_process_memory is %d KB",
                                  globalMemoryKb, localMemoryKb, maxReserveMemoryKb, g_instance.attr.attr_memory.max_process_memory);
                break;
            }
        }

        if (!htapEngine->Initialize(false)) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to initialize htap engine");
            break;
        }

        // make sure current thread is cleaned up properly when thread pool is enabled
        // avoid throwing errors on failure
        if (!htap::HTAPEnsureSafeThreadAccess()) {
            HTAP_REPORT_ERROR(HTAP_ERROR_INTERNAL, "Init HTAP", "Failed to initialize thread-local data");
            break;
        }

        htap::HTAPInitSessionDetailsMap();
        if (!g_instance.attr.attr_common.enable_thread_pool) {
            htap::HTAPInitSessionCleanup();
        }

        HTAP_LOG_INFO("Switching to external snapshot manager");
        htapEngine->SetCSNManager(&g_snapshotMgr);
        HTAP_LOG_INFO("htap engine initialized successfully");
        htapToTerminate = true;
        return true;
    } while (false);

    // cleanup
    if (addedCfgListener) {
        (void)htapEngine->RemoveConfigLoader(htapConfigLoader);
    }
    if (htapConfigLoader != nullptr) {
        delete htapConfigLoader;
        htapConfigLoader = nullptr;
    }
    if (htapEngine != nullptr) {
        MOT::MOTEngine::DestroyInstance();
        htapEngine = nullptr;
    }

    HTAP_LOG_ERROR("Failed to initialize  htap engine");
    return false;
}

void TermHTAPEngine()
{
    HTAP_LOG_INFO("Terminating htap engine");
    if (htapEngine == nullptr) {
        HTAP_LOG_WARN("htap engine is null");
    } else if (!htapToTerminate) {
        HTAP_LOG_INFO("htap engine was initialized externally, therefore will not be terminated by HTAP");
        htapEngine = nullptr;
    } else {
        htap::HTAPDestroySessionDetailsMap();
        if (htapConfigLoader != nullptr) {
            (void)htapEngine->RemoveConfigLoader(htapConfigLoader);
            delete htapConfigLoader;
            htapConfigLoader = nullptr;
        }
        
        // avoid throwing errors and ignore them at this phase
        (void)htap::HTAPEnsureSafeThreadAccess();
        // remove tables from memory from memcheck to pass
        MOT::MOTEngine::GetInstance()->GetTableManager()->ClearAllTables();
        MOT::MOTEngine::DestroyInstance();
        htapEngine = nullptr;
        // NOTE: if re-initialization is possible, then we should also reset all thread-locals
        // currently there is no such need
    }
}

void HTAPAdaptor::Init()
{
    // guard against double initialization (not thread safe)
    MOT::SetLogModuleName("HTAP");
    MOT::SetLogModuleId(MOD_HTAP);
    MOT::SetErrorModuleInfo(MOD_HTAP, "HTAP", HTAPErrorCodeTranslate);
    HTAP_LOG_INFO("Initializing HTAP system");
    if (m_initialized) {
        // This is highly unexpected, and should especially be guarded in scenario of switch-over to standby.
        HTAP_LOG_WARN("Double attempt to initialize HTAP system, it is already initialized");
        return;
    }

    if (!InitHTAPEngine()) {
        HTAP_LOG_ERROR_STACK("Failed to initialize memory store");
        HTAP_LOG_ERROR_EX("Failed to initialize memory store");
        return;
    }

    // create the HTAP facade
    if (!htap::HTAPInfra::CreateInstance(g_instance.attr.attr_common.HTAPConfigFileName)) {
        HTAP_LOG_ERROR_STACK("Failed to initialize HTAP system");
        HTAP_LOG_ERROR_EX("Failed to initialize HTAP system");
        return;
    }

    HTAP_LOG_INFO("HTAP system initialized successfully");
    m_initialized = true;
}

void HTAPAdaptor::Destroy()
{
    if (m_initialized) {
        HTAP_LOG_INFO("Terminating HTAP system");
        htap::HTAPInfra::DestroyInstance();
        TermHTAPEngine();
        HTAP_LOG_INFO("HTAP system terminated successfully");
        MOT::ClearLogModuleName();
        m_initialized = false;
    }
}

void HTAPAdaptor::DatumToMOT(MOT::Column *col, Datum datum, Oid type, uint8_t *data)
{
    switch (type) {
        case BYTEAOID:
        case TEXTOID:
        case VARCHAROID:
        case CLOBOID:
        case NVARCHAR2OID:
        case BPCHAROID: {
            bytea *txt = DatumGetByteaP(datum);
            size_t size = VARSIZE(txt); // includes header len VARHDRSZ
            char *src = VARDATA(txt);
            MOT::VlColumnHdr_t *hdr = static_cast<MOT::VlColumnHdr_t *>(static_cast<void *>(data + col->m_offset));
            HTAP_LOG_DEBUG("Column type: %s, size: %u, hdr size: %u, col->m_offset %u",
                           MOT::Column::ColumnTypeToStr(col->m_type), size - VARHDRSZ, hdr->size, col->m_offset);

            if (!col->Pack(data, (uintptr_t)src, size - VARHDRSZ)) {
                ereport(ERROR, (errcode(ERRCODE_STRING_DATA_RIGHT_TRUNCATION),
                                errmsg("value too long for column size (%d)", (int)(col->m_size - VARHDRSZ))));
            }
            break;
        }
        case NUMERICOID: {
            Numeric n = DatumGetNumeric(datum);
            char buf[DECIMAL_MAX_SIZE];
            MOT::DecimalSt *d = (MOT::DecimalSt *)buf;

            if (NUMERIC_NDIGITS(n) > DECIMAL_MAX_DIGITS) {
                ereport(ERROR, (errmodule(MOD_MOT), errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
                                errmsg("Value exceeds maximum precision: %d", NUMERIC_MAX_PRECISION)));
                break;
            }
            PGNumericToMOT(n, *d);
            (void)col->Pack(data, (uintptr_t)d, DECIMAL_SIZE(d)); // simple types packing cannot fail
            break;
        }
        default:
            (void)col->Pack(data, datum, col->m_size); // no verification required
            break;
    }
}

MOT::RC HTAPAdaptor::PackRow(TupleTableSlot *slot, MOT::Table *table, uint8_t *attrs_used, uint8_t *destRow)
{
    TupleDesc tupdesc = slot->tts_tupleDescriptor;
    uint64_t i = 0;
    uint8_t *bits = destRow + table->GetFieldOffset(i);

    // we now copy the not-null fields and update the null bits
    for (; i < table->GetNumIMCSColumns(); i++) {
        bool isnull = false;

        int2 imcs_key = table->GetIMCSColumnKey(i);
        if (imcs_key == -1) {
            HTAP_LOG_ERROR("IMCS key is absent");
            return MOT::RC_ERROR;
        }

        Datum value = tableam_tslot_getattr(slot, imcs_key, &isnull);
        if (!isnull) {
            DatumToMOT(table->GetField(i + 1), value, tupdesc->attrs[imcs_key - 1].atttypid, destRow);
            BITMAP_SET(bits, i);
        } else {
            BITMAP_CLEAR(bits, i);
        }
    }

    return MOT::RC_OK;
}

static MOT::RC PackRowParsed(TupleTableSlot *slot, MOT::Table *table, uint8_t *attrs_used, uint8_t *destRow,
    uint32_t varRowSize, uint32_t *columnExtSize, Datum *values, bool *nulls)
{
    TupleDesc tupdesc = slot->tts_tupleDescriptor;
    uint64_t i = 0;
    uint8_t *bits = destRow + table->GetFieldOffset(i);
    uint32_t tableTupleSize = ALIGN8(table->GetTupleSize());
    uint32_t varLenOffset = tableTupleSize;
    // we now copy the not-null fields and update the null bits
    for (; i < table->GetNumIMCSColumns(); i++) {
        int2 imcs_key = table->GetIMCSColumnKey(i);
        if (imcs_key == -1) {
            HTAP_LOG_ERROR("IMCS key is absent");
            return MOT::RC_ERROR;
        }
        MOT::Column *col = table->GetField(i + 1);
        if (MOT::isVariableLengthMotType(col->m_type)) {
            // update field size
            MOT::VlColumnHdr_t *hdr = reinterpret_cast<MOT::VlColumnHdr_t *>(((char *)destRow) + col->m_offset);
            hdr->size = columnExtSize[i+1];
            hdr->offset = hdr->size == 0 ? 0 : varLenOffset;
            varLenOffset += ALIGN8(hdr->size);
        }

        if (!nulls[i]) {
            HTAPAdaptor::DatumToMOT(col, values[i], tupdesc->attrs[imcs_key - 1].atttypid, destRow);
            BITMAP_SET(bits, i);
        } else {
            BITMAP_CLEAR(bits, i);
        }
    }

    return MOT::RC_OK;
}

static MOT::RC TableFieldType(MOT::MOT_CATALOG_FIELD_TYPES &type, Oid typoid, int typeLen, bool supportVariableLength)
{
    MOT::RC res = MOT::RC_OK;

    switch (typoid) {
        case CHAROID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_CHAR;
            break;
        case INT1OID:
        case BOOLOID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_TINY;
            break;
        case INT2OID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_SHORT;
            break;
        case INT4OID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_INT;
            break;
        case INT8OID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_LONG;
            break;
        case DATEOID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_DATE;
            break;
        case TIMEOID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_TIME;
            break;
        case TIMESTAMPOID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_TIMESTAMP;
            break;
        case TIMESTAMPTZOID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_TIMESTAMPTZ;
            break;
        case INTERVALOID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_INTERVAL;
            break;
        case TINTERVALOID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_TINTERVAL;
            break;
        case TIMETZOID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_TIMETZ;
            break;
        case FLOAT4OID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_FLOAT;
            break;
        case FLOAT8OID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_DOUBLE;
            break;
        case NUMERICOID:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_DECIMAL;
            break;
        case VARCHAROID:
        case BPCHAROID:
        case BYTEAOID:
        case NVARCHAR2OID:
        case TEXTOID:
        case CLOBOID:
            // variable-length row has overhead. avoid varaible-length type for short-length types
            if (supportVariableLength && (typeLen >= MOT::Table::VL_TYPE_MIN_SIZE)) {
                type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_HTAPBLOB;
            } else {
                type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_VARCHAR;
            }
            break;
        default:
            type = MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_UNKNOWN;
            res = MOT::RC_UNSUPPORTED_COL_TYPE;
    }

    return res;
}

static int2 GetAttSize(MOT::MOT_CATALOG_FIELD_TYPES type, Form_pg_attribute att_tup)
{
    int2 len = 0;
    switch (type) {
        case MOT::MOT_TYPE_HTAPBLOB:
            len = sizeof(MOT::VlColumnHdr_t);
            break;
        case MOT::MOT_TYPE_VARCHAR: {
            if (att_tup->atttypmod != -1) {
                len = att_tup->atttypmod;
            } else {
                if (att_tup->atttypid == TEXTOID) {
                    len = MAX_VARCHAR_LEN;
                } else {
                    len = DEFAULT_VARCHAR_LEN;
                }
            }
            len += VARHDRSZ;
        } break;

        case MOT::MOT_TYPE_DECIMAL:
            if (att_tup->atttypmod == -1)
                len = DECIMAL_MAX_SIZE;
            else
                len = (((att_tup->atttypmod >> 16) + (att_tup->atttypmod & 0xFFFF)) * sizeof(NumericDigit));
            break;

        default:
            len = att_tup->attlen;
            break;
    }
    return len;
}

bool HTAPAdaptor::IsTxnWriteSetEmpty()
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    if (txn->m_txnDdlAccess->Size() > 0 or txn->m_accessMgr->Size() > 0) {
        return false;
    }
    return true;
}

MOT::RC HTAPAdaptor::ValidateCommit()
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    if (!IS_PGXC_COORDINATOR) {
        return txn->ValidateCommit();
    } else {
        // Nothing to do in coordinator
        return MOT::RC_OK;
    }
}

void HTAPAdaptor::RecordCommit(uint64_t csn)
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    txn->SetCommitSequenceNumber(csn);
    if (!IS_PGXC_COORDINATOR) {
        txn->RecordCommit();
    } else {
        txn->LiteCommit();
    }
}

MOT::RC HTAPAdaptor::Commit(uint64_t csn)
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    txn->SetCommitSequenceNumber(csn);
    if (!IS_PGXC_COORDINATOR) {
        return txn->Commit();
    } else {
        txn->LiteCommit();
        return MOT::RC_OK;
    }
}

void HTAPAdaptor::EndTransaction()
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    // Nothing to do in coordinator
    if (!IS_PGXC_COORDINATOR) {
        txn->EndTransaction();
    }
}

void HTAPAdaptor::Rollback()
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    txn->ClearOidNodeInf();
    if (!IS_PGXC_COORDINATOR) {
        txn->Rollback();
    } else {
        txn->LiteRollback();
    }
}

MOT::RC HTAPAdaptor::Prepare()
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    if (!IS_PGXC_COORDINATOR) {
        return txn->Prepare();
    } else {
        txn->LitePrepare();
        return MOT::RC_OK;
    }
}

void HTAPAdaptor::CommitPrepared(uint64_t csn)
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    txn->SetCommitSequenceNumber(csn);
    if (!IS_PGXC_COORDINATOR) {
        txn->CommitPrepared();
    } else {
        txn->LiteCommitPrepared();
    }
}

void HTAPAdaptor::RollbackPrepared()
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    if (!IS_PGXC_COORDINATOR) {
        txn->RollbackPrepared();
    } else {
        txn->LiteRollbackPrepared();
    }
}

RC HTAPAdaptor::DropTable(Oid rel_oid, std::unordered_map<Oid, RelFileNode> **out_oid_del_info)
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    if (txn == nullptr) {
        HTAP_REPORT_ERROR(HTAP_RC_INVALID_STATE, "Drop IMCStore delta Table",
            "Cannot drop IMCStore delta table %u: No current transaction is defined", rel_oid);
        return HTAP_RC_INVALID_STATE;
    }

    txn->SetTransactionId(GetCurrentTransactionIdIfAny());
    MOT::Table *table = txn->GetTableByExternalId(rel_oid);
    if (table == nullptr) {
        HTAP_LOG_TRACE("No IMCStore delta table found for relation %u", rel_oid);
        return HTAP_RC_TABLE_NOT_FOUND;
    }

    RC res = HTAP_RC_OK;
    table->GetOrigTable()->WrLock();
    MOT::RC motRes = txn->DropTable(table);
    table->GetOrigTable()->Unlock();
    if (motRes != MOT::RC_OK) {
        res = ToHTAPRc(motRes);
        HTAP_REPORT_ERROR(res, "DropTable", "Failed to drop IMCStore delta table for relation %u: %s", rel_oid,
            MOT::RcToString(motRes));
    }

    HTAP_LOG_INFO("Dropped table %s with ID %u", table->GetTableName().c_str(), rel_oid);

    if (IMCU_CACHE->m_reinit_imcs_set.count(rel_oid) != 0) {
        (void)GetTableManager()->DropTable((table), txn);
        return res;
    }

    if (out_oid_del_info)
        *out_oid_del_info = txn->GetOidNodeInf(rel_oid);

    return res;
}

RC HTAPAdaptor::CreateTable(Oid parent_oid, Oid rel_oid)
{
    MOT::RC motRes = MOT::RC_OK;
    Relation htap_rel = NULL;
    MOT::Index *primaryIdx = nullptr;
    MOT::TxnManager *txn;
    MOT::Table *table = nullptr;
    uint32_t columnCount;
    int2vector *imcs_keys = nullptr;
    uint32_t num_of_imcs_columns;
    int2 *column_keys = nullptr;
    bool txnTableCreated = false;
    errno_t rc = EOK;
    char *dbname = NULL;
    std::string tname("");

    HTAP_LOG_INFO("Start create table with ID %u ", rel_oid);
    char *imcs_delta_name = (char *)palloc0(sizeof(char) * NAMEDATALEN);
    rc = snprintf_s(imcs_delta_name, NAMEDATALEN, NAMEDATALEN - 1, "gs_imcsdelta_%u", rel_oid);
    securec_check_ss(rc, "", "");

    imcs_keys = GetImcsKeys(rel_oid);
    if (!imcs_keys) {
        HTAP_REPORT_ERROR(HTAP_RC_INTERNAL_ERROR, "Create IMCStore delta Table", "NULL imcstored columns in table %u", rel_oid);
        return HTAP_RC_INTERNAL_ERROR;
    }

    num_of_imcs_columns = imcs_keys->dim1;
    if (num_of_imcs_columns == 0) {
        HTAP_REPORT_ERROR(HTAP_RC_INVALID_ARG, "Create IMCStore delta Table", "No imcstored columns in the table %u", rel_oid);
        return HTAP_RC_INVALID_ARG;
    }

    columnCount = num_of_imcs_columns;

    txn = HTAPGetSafeTxn();
    if (txn == nullptr) {
        HTAP_REPORT_ERROR(GetRootError(), "Create IMCStore delta Table", "Failed to initialize current transaction");
        return GetRootError();
    }

    txn->SetTransactionId(GetCurrentTransactionIdIfAny());
    HTAP_LOG_INFO("Start create table for parent: %d, rel: %d", parent_oid, rel_oid);
    htap_rel = heap_open(parent_oid, NoLock);

    RC res = HTAP_RC_OK;
    do {
        const char *prevDBName = MOT::GetTableManager()->GetFirstDbName();
        dbname = get_database_name(u_sess->proc_cxt.MyDatabaseId);
        if (dbname == nullptr) {
            delete table;
            table = nullptr;
            ereport(ERROR, (errcode(ERRCODE_UNDEFINED_DATABASE),
                errmsg("database with OID %u does not exist", u_sess->proc_cxt.MyDatabaseId)));
            res = HTAP_RC_ERROR;
            break;
        } else if (prevDBName != nullptr && strcasecmp(dbname, prevDBName)) {
            delete table;
            table = nullptr;
            ereport(ERROR, (errcode(ERRCODE_UNDEFINED_DATABASE), errmsg(
                "Can not support IMCSTORED tables in multiple databases prev [%s] new [%s]", prevDBName, dbname)));
            res = HTAP_RC_ERROR;
            break;
        }
        tname.append(imcs_delta_name);
        tname.append("_");
        tname.append(htap_rel->rd_rel->relname.data);
        table = new (std::nothrow) MOT::Table(true);
        if (table == nullptr) {
            HTAP_REPORT_ERROR(HTAP_RC_OOM, "Create IMCStore delta Table", "Failed to allocate memory for table");
            res = HTAP_RC_OOM;
            break;
        }

        // once the columns have been counted, we add one more for the nullable columns
        ++columnCount;

        if (!table->Init(tname.c_str(), tname.c_str(), columnCount, rel_oid)) {
            HTAP_REPORT_ERROR(HTAP_RC_INTERNAL_ERROR, "Create IMCStore delta Table",
                "Failed to initialize HTAP IMCStore delta table %s (%u)", tname.c_str(), rel_oid);
            res = HTAP_RC_INTERNAL_ERROR;
            break;
        }
        table->SetDbName(dbname);
        column_keys = (int2 *)malloc(2 * num_of_imcs_columns);
        if (column_keys == nullptr) {
            HTAP_REPORT_ERROR(HTAP_RC_OOM, "Create IMCStore delta Table",
                "Failed to allocate %u bytes for column_keys of IMCStore delta table %s (%u)", (2 * num_of_imcs_columns),
                tname.c_str(), rel_oid);
            res = HTAP_RC_OOM;
            break;
        }

        for (uint32_t i = 0; i < num_of_imcs_columns; i++) {
            column_keys[i] = imcs_keys->values[i];
        }

        table->SetIMCSAttributes(column_keys, num_of_imcs_columns);
        MOT::MOT_CATALOG_FIELD_TYPES type;

        // the null fields are copied verbatim because we have to give them back at some point
        motRes = table->AddColumn("null_bytes", BITMAPLEN(columnCount - 1),
            MOT::MOT_CATALOG_FIELD_TYPES::MOT_TYPE_NULLBYTES);
        if (motRes != MOT::RC_OK) {
            res = ToHTAPRc(motRes);
            HTAP_REPORT_ERROR(res, "Create IMCStore delta Table", "Failed to add null_bytes column to IMCStore delta table %s (%u): %s",
                tname.c_str(), rel_oid, MOT::RcToString(motRes));
            break;
        }

        for (uint32_t i = 0; i < num_of_imcs_columns; i++) {
            AttrNumber attnum = column_keys[i];
            Form_pg_attribute att_tup = &htap_rel->rd_att->attrs[attnum - 1];
            motRes = TableFieldType(type, att_tup->atttypid, att_tup->atttypmod, table->IsSupportVariableLength());
            if (motRes != MOT::RC_OK) {
                res = ToHTAPRc(motRes);
                HTAP_REPORT_ERROR(res, "Create IMCStore delta Table",
                    "Cannot add column with unsupported type to IMCStore delta table %s (%u): (%s)", tname.c_str(), rel_oid,
                    MOT::RcToString(motRes));
                break;
            }

            motRes = table->AddColumn(att_tup->attname.data, GetAttSize(type, att_tup), type, att_tup->attnotnull,
                att_tup->atttypid);

            if (motRes != MOT::RC_OK) {
                res = ToHTAPRc(motRes);
                HTAP_REPORT_ERROR(res, "Create IMCStore delta Table", "Failed to add column to IMCStore delta table %s (%u): (%s)",
                    tname.c_str(), rel_oid, MOT::RcToString(motRes));
                break;
            }

            if (isVariableLengthMotType(type) && table->IsFixedLengthRow()) {
                table->SetFixedLengthRow(false);
            }
        }

        if (res != HTAP_RC_OK) {
            break;
        }

        if (table->IsFixedLengthRow()) {
            if (!table->InitRowPool()) {
                HTAP_REPORT_ERROR(HTAP_RC_OOM, "Create IMCStore delta Table",
                    "Failed to initialize row pool of IMCStore delta table %s (%u)", tname.c_str(), rel_oid);
                res = HTAP_RC_OOM;
                break;
            }
        } else {
            if (!table->InitRowSlab()) {
                HTAP_REPORT_ERROR(HTAP_RC_OOM, "Create IMCStore delta Table",
                    "Failed to initialize row slab of IMCStore delta table %s (%u)", tname.c_str(), rel_oid);
                res = HTAP_RC_OOM;
                break;
            }
        }

        if (!table->InitTombStonePool()) {
            HTAP_REPORT_ERROR(HTAP_RC_INTERNAL_ERROR, "Create IMCStore delta Table",
                "Failed to initialize tombstone pool of IMCStore delta table %s (%u)", tname.c_str(), rel_oid);
            res = HTAP_RC_INTERNAL_ERROR;
            break;
        }

        motRes = txn->CreateTable(table);
        if (motRes != MOT::RC_OK) {
            HTAP_REPORT_ERROR(HTAP_RC_INTERNAL_ERROR, "Create IMCStore delta Table",
                "Failed to create transactional IMCStore delta table %s (%u)", tname.c_str(), rel_oid);
            break;
        }

        txnTableCreated = true;

        // add default PK index
        primaryIdx = MOT::IndexFactory::CreatePrimaryIndexEx(MOT::IndexingMethod::INDEXING_METHOD_TREE,
            DEFAULT_TREE_FLAVOR, sizeof(uint64_t), table->GetLongTableName(), motRes, nullptr);

        if (motRes != MOT::RC_OK) {
            res = ToHTAPRc(motRes);
            HTAP_REPORT_ERROR(res, "Create IMCStore delta Table", "Failed to create primary index for IMCStore delta table %s (%u): %s",
                tname.c_str(), rel_oid, MOT::RcToString(motRes));
            break;
        }

        if (!primaryIdx->SetNumTableFields(columnCount)) {
            HTAP_REPORT_ERROR(HTAP_RC_INTERNAL_ERROR, "Create IMCStore delta Table",
                "Failed to create primary index bitmap for IMCStore delta table %s (%u)", tname.c_str(), rel_oid);
            break;
        }
        primaryIdx->SetNumIndexFields(1);
        primaryIdx->SetLenghtKeyFields(0, -1, sizeof(uint64_t));
        primaryIdx->SetCTIDPrimary(true);
        primaryIdx->SetExtId(rel_oid);
        // Add default primary index
        motRes = txn->CreateIndex(table, primaryIdx, true);
        if (motRes != MOT::RC_OK) {
            res = ToHTAPRc(motRes);
            HTAP_REPORT_ERROR(res, "Create IMCStore delta Table",
                "Failed to create transactional primary index for IMCStore delta table %s (%u): %s", tname.c_str(), rel_oid,
                MOT::RcToString(motRes));
            break;
        }
    } while (0);

    heap_close(htap_rel, NoLock);

    if (res != HTAP_RC_OK) {
        if (primaryIdx != nullptr) {
            delete primaryIdx;
            primaryIdx = nullptr;
        }

        if (txnTableCreated) {
            txn->DropTable(table);
        } else if (table != nullptr) {
            delete table;
            table = nullptr;
        }
    } else {
        HTAP_LOG_INFO("Created table %s with ID %u", tname.c_str(), rel_oid);
    }

    return res;
}

RC HTAPAdaptor::TableModify(Oid rel_oid, ItemPointer ctid, TupleTableSlot *slot)
{
    MOT::RC motRes = MOT::RC_OK;
    uint8_t *newRowData = nullptr;
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    MOT::Table *table;
    MOT::Index *index;
    MOT::Row *row = nullptr;
    MOT::Sentinel *s = nullptr;
    MOT::Row *currRow = nullptr;
    HTAPCTID rowCtid(ctid->ip_blkid.bi_hi, ctid->ip_blkid.bi_lo, ctid->ip_posid, HTAP_DELTA_UPDATE_TAG);
    uint64_t flatCTID = 0;
    MOT::MaxKey htap_primary_key;

    table = txn->GetTableByExternalId(rel_oid);
    if (table == nullptr) {
        HTAP_REPORT_ERROR(HTAP_RC_TABLE_NOT_FOUND, "TableModify", "Failed to find IMCStore delta table by external id: %u",
            rel_oid);
        return HTAP_RC_TABLE_NOT_FOUND;
    }

    int numImcsCols = table->GetNumIMCSColumns() + 1;
    Datum values[numImcsCols];
    bool nulls[numImcsCols];
    uint32_t columnExtSize[numImcsCols] = { 0 };
    uint32_t varRowSize = 0;
    txn->SetTransactionId(GetCurrentTransactionId());

    flatCTID = rowCtid.GetCTIDKey();
    index = table->GetPrimaryIndex();
    htap_primary_key.InitKey(index->GetKeyLength());
    htap_primary_key.CpKey((uint8_t *)(&flatCTID), index->GetKeyLength());

    s = index->IndexReadSentinel(&htap_primary_key, txn->GetThdId());
    if (s) {
        motRes = s->RefCountUpdate(MOT::AccessType::INC);
        if (motRes == MOT::RC_OK) {
            currRow = txn->RowLookup(MOT::AccessType::WR, s, motRes);
            if (motRes != MOT::RC_OK) {
                RC res = ToHTAPRc(motRes);
                s->RefCountUpdate(MOT::AccessType::DEC);
                HTAP_REPORT_ERROR(res, "TableModify", "Failed to lookup row for table %s (id: %u): %s",
                    table->GetTableName().c_str(), rel_oid, MOT::RcToString(motRes));
                return res;
            }
        }
    }
    if (currRow != nullptr) {
        varRowSize = table->CalcVarRowSize(slot, columnExtSize, values, nulls);
        motRes = txn->UpdateLastRowState(MOT::AccessType::WR, varRowSize);
        if (motRes != MOT::RC_OK) {
            RC res = ToHTAPRc(motRes);
            s->RefCountUpdate(MOT::AccessType::DEC);
            HTAP_REPORT_ERROR(res, "TableModify", "Failed to update row state for table %s (id: %u): %s",
                table->GetTableName().c_str(), rel_oid, MOT::RcToString(motRes));
            return res;
        }
        row = txn->GetLastAccessedDraft();
        if (row->GetCTIDStatus() != HTAP_DELTA_INSERT_TAG) {
            row->SetCTIDKey(rowCtid.m_u.m_ctid);
        }
        row->SetKeytype(MOT::KeyType::CTID_KEY);

        if (!table->IsFixedLengthRow()) {
            /* optimization: For HTAPBLOB we have few options:
                1. Each new value can fit in the same original field
                2. If not:
                    Current row size might be enough to all fields, but we need to rebuild it (change size of each
            field) If not (new Row's fields require more space than allocated): Allocate new row from given slot

            We currently allocate new row for all cases. We need to optimize this code
            */
            if (varRowSize > row->GetAllocSize()) {
                MOT::Row *newRow = table->CreateNewRowCopy(row, MOT::AccessType::WR, varRowSize);
                if (newRow == nullptr) {
                    s->RefCountUpdate(MOT::AccessType::DEC);
                    HTAP_REPORT_ERROR(HTAP_RC_OOM, "TableModify", "Failed to allocate row for table %s (id: %u)",
                        table->GetTableName().c_str(), rel_oid);
                    return HTAP_RC_OOM;
                }
                // replace last accessed draft
                txn->SetLastAccessedDraft(newRow);
                row->GetTable()->DestroyRow(row);
                row = txn->GetLastAccessedDraft();
            }
            varRowSize = row->GetAllocSize();
            newRowData = const_cast<uint8_t *>(row->GetData());
            motRes = PackRowParsed(slot, table, 0, newRowData, varRowSize, columnExtSize, values, nulls);
        } else {
            newRowData = const_cast<uint8_t *>(row->GetData());
            motRes = PackRow(slot, table, 0, newRowData);
        }
        if (motRes != MOT::RC_OK) {
            RC res = ToHTAPRc(motRes);
            s->RefCountUpdate(MOT::AccessType::DEC);
            HTAP_REPORT_ERROR(res, "TableModify", "Failed to pack row for table %s (id: %u): %s",
                table->GetTableName().c_str(), rel_oid, MOT::RcToString(motRes));
            return res;
        }
        txn->OverwriteRowHtap(row);
    } else {
        if (s) {
            MOT_ASSERT(s->GetCounter() == 0);
        }
        row = table->CreateNewRow(slot, varRowSize, columnExtSize, values, nulls);
        if (row == nullptr) {
            HTAP_REPORT_ERROR(HTAP_RC_OOM, "TableModify", "Failed to allocate row for table %s (id: %u)",
                table->GetTableName().c_str(), rel_oid);
            return HTAP_RC_OOM;
        }
        row->SetNextVersion(nullptr);
        row->SetCTIDKey(rowCtid.m_u.m_ctid);
        row->SetKeytype(MOT::KeyType::CTID_KEY);
        newRowData = const_cast<uint8_t *>(row->GetData());

        if (!table->IsFixedLengthRow()) {
            motRes =
                PackRowParsed(slot, table->GetOrigTable(), 0, newRowData, varRowSize, columnExtSize, values, nulls);
        } else {
            motRes = PackRow(slot, table->GetOrigTable(), 0, newRowData);
        }
        if (motRes != MOT::RC_OK) {
            RC res = ToHTAPRc(motRes);
            HTAP_REPORT_ERROR(res, "TableModify", "Failed to pack row for table %s (id: %u): %s",
                table->GetTableName().c_str(), rel_oid, MOT::RcToString(motRes));
            table->DestroyRow(row);
            return res;
        }

        motRes = table->InsertRow(row, txn);
        if (motRes != MOT::RC_OK) {
            RC res = ToHTAPRc(motRes);
            HTAP_REPORT_ERROR(res, "TableModify", "Failed to insert row into table %s (id: %u): %s",
                table->GetTableName().c_str(), rel_oid, MOT::RcToString(motRes));
            return res;
        }
    }
    return HTAP_RC_OK;
}

RC HTAPAdaptor::TableInsert(Oid rel_oid, ItemPointer ctid, TupleTableSlot *slot)
{
    MOT::RC motRes = MOT::RC_OK;
    uint8_t *newRowData = nullptr;
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    MOT::Table *table;
    MOT::Row *row = nullptr;
    HTAPCTID rowCtid(ctid->ip_blkid.bi_hi, ctid->ip_blkid.bi_lo, ctid->ip_posid, HTAP_DELTA_INSERT_TAG);

    table = txn->GetTableByExternalId(rel_oid);
    if (table == nullptr) {
        HTAP_REPORT_ERROR(HTAP_RC_TABLE_NOT_FOUND, "TableInsert", "Failed to find IMCStore delta table by external id: %u",
            rel_oid);
        return HTAP_RC_TABLE_NOT_FOUND;
    }

    int numImcsCols = table->GetNumIMCSColumns() + 1;
    Datum values[numImcsCols];
    bool nulls[numImcsCols];
    uint32_t columnExtSize[numImcsCols] = { 0 };
    uint32_t varRowSize = 0;

    row = table->CreateNewRow(slot, varRowSize, columnExtSize, values, nulls);
    if (row == nullptr) {
        HTAP_REPORT_ERROR(HTAP_RC_OOM, "TableInsert", "Failed to allocate row for table %s (id: %u)",
            table->GetTableName().c_str(), rel_oid);
        return HTAP_RC_OOM;
    }
    row->SetNextVersion(nullptr);
    row->SetCTIDKey(rowCtid.m_u.m_ctid);
    row->SetKeytype(MOT::KeyType::CTID_KEY);
    newRowData = const_cast<uint8_t *>(row->GetData());

    if (!table->IsFixedLengthRow()) {
        motRes = PackRowParsed(slot, table, 0, newRowData, varRowSize, columnExtSize, values, nulls);
    } else {
        motRes = PackRow(slot, table, 0, newRowData);
    }
    if (motRes != MOT::RC_OK) {
        RC res = ToHTAPRc(motRes);
        HTAP_REPORT_ERROR(res, "TableInsert", "Failed to pack row for table %s (id: %u): %s",
            table->GetTableName().c_str(), rel_oid, MOT::RcToString(motRes));
        table->DestroyRow(row);
        return res;
    }

    txn->SetTransactionId(GetCurrentTransactionIdIfAny());
    motRes = table->InsertRow(row, txn);
    if (motRes != MOT::RC_OK) {
        RC res = ToHTAPRc(motRes);
        HTAP_REPORT_ERROR(res, "TableInsert", "Failed to insert row into table %s (id: %u): %s",
            table->GetTableName().c_str(), rel_oid, MOT::RcToString(motRes));
        return res;
    }

    return HTAP_RC_OK;
}

RC HTAPAdaptor::TableDelete(Oid rel_oid, ItemPointer ctid, bool fromUpd)
{
    MOT::RC motRes = MOT::RC_OK;
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    MOT::Table *table;
    MOT::Index *index;
    MOT::Row *row = nullptr;
    MOT::Sentinel *s = nullptr;
    MOT::Row *currRow = nullptr;
    HTAPCTID rowCtid(ctid->ip_blkid.bi_hi, ctid->ip_blkid.bi_lo, ctid->ip_posid, HTAP_DELTA_DELETE_TAG);
    uint64_t flatCTID = 0;
    MOT::MaxKey htap_primary_key;

    table = txn->GetTableByExternalId(rel_oid);
    if (table == nullptr) {
        HTAP_REPORT_ERROR(HTAP_RC_TABLE_NOT_FOUND, "TableDelete", "Failed to find IMCStore delta table by external id: %u",
            rel_oid);
        return HTAP_RC_TABLE_NOT_FOUND;
    }

    flatCTID = rowCtid.GetCTIDKey();
    index = table->GetPrimaryIndex();
    htap_primary_key.InitKey(index->GetKeyLength());
    htap_primary_key.CpKey((uint8_t *)(&flatCTID), index->GetKeyLength());
    txn->SetTransactionId(GetCurrentTransactionIdIfAny());

    s = index->IndexReadSentinel(&htap_primary_key, txn->GetThdId());
    if (s) {
        motRes = s->RefCountUpdate(MOT::AccessType::INC);
        if (motRes == MOT::RC::RC_OK) {
            currRow = txn->RowLookup((fromUpd ? MOT::AccessType::WR : MOT::AccessType::DEL), s, motRes);
            if (motRes != MOT::RC_OK) {
                RC res = ToHTAPRc(motRes);
                s->RefCountUpdate(MOT::AccessType::DEC);
                HTAP_REPORT_ERROR(res, "TableDelete", "Failed to lookup row in table %s (id: %u): %s",
                    table->GetTableName().c_str(), rel_oid, MOT::RcToString(motRes));
                return res;
            }
        }
    }
    if (currRow != nullptr) {
        motRes = txn->DeleteLastRow();
        if (motRes != MOT::RC_OK) {
            RC res = ToHTAPRc(motRes);
            s->RefCountUpdate(MOT::AccessType::DEC);
            HTAP_REPORT_ERROR(res, "TableDelete", "Failed to delete row from table %s (id: %u): %s",
                table->GetTableName().c_str(), rel_oid, MOT::RcToString(motRes));
            return res;
        }
        row = txn->GetLastAccessedDraft();
        if (row != nullptr) {
            row->SetCTIDKey(rowCtid.m_u.m_ctid);
        }
        return HTAP_RC_OK;
    } else {
        if (s) {
            MOT_ASSERT(s->GetCounter() == 0);
        }
        row = table->CreateTombStone();
        if (row == nullptr) {
            HTAP_REPORT_ERROR(HTAP_RC_OOM, "TableDelete", "Failed to allocate row for table %s (id: %u)",
                table->GetTableName().c_str(), rel_oid);
            return HTAP_RC_OOM;
        }
        row->SetNextVersion(nullptr);
        row->SetCTIDKey(rowCtid.m_u.m_ctid);
        row->SetKeytype(MOT::KeyType::CTID_KEY);
        motRes = table->InsertRow(row, txn);
        if (motRes != MOT::RC_OK) {
            RC res = ToHTAPRc(motRes);
            HTAP_REPORT_ERROR(res, "TableDelete", "Failed to insert delete row into table %s (id: %u): %s",
                table->GetTableName().c_str(), rel_oid, MOT::RcToString(motRes));
            return res;
        }
    }
    return HTAP_RC_OK;
}

RC HTAPAdaptor::QueryDelta(Oid relationId, uint64_t &rowCount, uint64_t &newRCount, uint64_t &memSize)
{
    MOT::Table *table;
    MOT::TxnManager *txn = HTAPGetSafeTxn();
    table = txn->GetTableByExternalId(relationId);
    uint64_t netSize;
    HTAPDelta *htap_delta = New(CurrentMemoryContext)HTAPDelta(true, false);

    if (table == nullptr) {
        HTAP_REPORT_ERROR(HTAP_RC_INTERNAL_ERROR, "Query Delta", "Failed to get the IMCStore delta table of relation %u",
            relationId);
        return HTAP_RC_INTERNAL_ERROR;
    }
    memSize = table->GetTableSize(netSize, rowCount);
    if (htap_delta->BuildIndexOnDeltaTable(relationId, MOT::GetCSNManager().GetGcEpoch())) {
        newRCount = htap_delta->addRecords();
    }

    return HTAP_RC_OK;
}

RC HTAPAdaptor::TruncateTable(Oid rel_oid, std::unordered_map<Oid, RelFileNode> **out_oid_del_info)
{
    MOT::Table *table;
    MOT::TxnManager *txn = HTAPGetSafeTxn();

    txn->SetTransactionId(GetCurrentTransactionIdIfAny());
    table = txn->GetTableByExternalId(rel_oid);
    if (table == nullptr) {
        HTAP_REPORT_ERROR(HTAP_RC_INTERNAL_ERROR, "Truncate Table", "Failed to get the IMCStore delta table of relation %u",
            rel_oid);
        return HTAP_RC_INTERNAL_ERROR;
    }

    table->GetOrigTable()->WrLock();
    MOT::RC res = txn->TruncateTable(table);
    table->GetOrigTable()->Unlock();
    if (res != MOT::RC_OK) {
        HTAP_REPORT_ERROR(HTAP_RC_OOM, "TableTruncate", "Not enough memory for truncate for table %s (id: %u)",
            table->GetTableName().c_str(), rel_oid);
        return HTAP_RC_OOM;
    }
    if (out_oid_del_info)
        *out_oid_del_info = txn->GetOidNodeInf(rel_oid);
    return HTAP_RC_OK;
}

MOT::Table *HTAPAdaptor::getCurrentMOTTable(Oid rel_oid)
{
    MOT::TxnManager *txn = HTAPGetSafeTxn();

    if (txn == nullptr) {
        HTAP_LOG_ERROR("HTAP::getMOTTable - Failed to get Txn Manager");
        return nullptr;
    }

    MOT::Table *table = txn->GetTableByExternalId(rel_oid);

    if (table == nullptr) {
        HTAP_LOG_ERROR("HTAP::getMOTTable - Failed to get MOT Table %u", rel_oid);
        return nullptr;
    }
    return table;
}
} // namespace htap

